/**
 * 
 */
package com.ffcs.crmd.tsp.core.netty.client;

import java.util.Iterator;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import org.apache.commons.lang3.StringUtils;

import com.ffcs.crmd.tsp.api.netty.INettyConfigClient;
import com.ffcs.crmd.tsp.api.netty.INettyProcessor;
import com.ffcs.crmd.tsp.api.netty.INettyRemotingClient;
import com.ffcs.crmd.tsp.core.netty.NettyRemotingAbstract;
import com.ffcs.crmd.tsp.core.netty.RpcPair;
import com.ffcs.crmd.tsp.core.thread.ThreadFactoryImpl;
import com.ffcs.crmd.tsp.core.util.RemotingUtil;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.util.concurrent.DefaultEventExecutorGroup;

/**
 * 功能说明:客户端netty通信抽象父类
 *
 * @author ZHONGFUHUA
 * 
 * @since 2.0.0-SNAPSHOT
 *
 */
public abstract class NettyRemotingClientAbstract extends NettyRemotingAbstract implements INettyRemotingClient {
    
    /**
     * I/O操作的多线程事件循环器
     */
    private final EventLoopGroup                           eventLoopGroupWorker;
    
    /**
     * I/O操作的多线程事件循环器
     */
    private final Bootstrap                                bootstrap                = new Bootstrap();
    
    /**
     * 通信服务端处理请求线程池
     */
    private final ExecutorService                          processsorExecutor;
    
    /**
     * 通信服务端处理请求应答线程池
     */
    private final ExecutorService                          callBackExecutor;
    
    /**
     * netty客户端配置
     */
    protected final INettyConfigClient                     nettyConfig;
    
    /**
     * 通道缓存锁超时时间
     */
    private long                                           lockTimeoutMillis        = 3000;
    
    /**
     * 通道缓存锁
     */
    private Lock                                           channelTableLock         = new ReentrantLock();
    
    /**
     * 通道缓存
     */
    private ConcurrentHashMap<String, NettyChannelWrapper> channelTables            = new ConcurrentHashMap<String, NettyChannelWrapper>();
    
    /**
     * 定时服务
     */
    private ScheduledExecutorService                       scheduledExecutorService = Executors
        .newSingleThreadScheduledExecutor(new ThreadFactoryImpl("TspClientNettyScheduledThread-"));
    
    /**
     * 
     */
    private DefaultEventExecutorGroup defaultGroup;
    
    /**
     * 
     * 功能说明:
     * 
     * @param nettyConfig
     */
    public NettyRemotingClientAbstract(INettyConfigClient nettyConfig) {
        super(nettyConfig.getPermitsOneway(), nettyConfig.getPermitsAsync());
        this.nettyConfig = nettyConfig;
        int callBackThreadNums = nettyConfig.getCallbackExecutorThreads();
        int processorThreadNums = nettyConfig.getProcessorExecutorThreads();
        if (callBackThreadNums <= 0) {
            callBackThreadNums = 4;
        }
        if (processorThreadNums <= 0) {
            processorThreadNums = 4;
        }
        
        callBackExecutor = Executors.newFixedThreadPool(callBackThreadNums, new ThreadFactoryImpl("TspClientNettyCallBackExecutor-"));
        processsorExecutor = Executors.newFixedThreadPool(processorThreadNums, new ThreadFactoryImpl("TspClientNettyProcesssorExecutor-"));
        
        eventLoopGroupWorker = new NioEventLoopGroup(1, new ThreadFactoryImpl("TspClientNettySelectorLoop-"));
        defaultGroup = new DefaultEventExecutorGroup(nettyConfig.getWorkGroupThreads(), new ThreadFactoryImpl("TspClientNettyWorkLoop-"));
    }
    
    public void start() {
        bootstrap.group(eventLoopGroupWorker).channel(NioSocketChannel.class)
            .option(ChannelOption.TCP_NODELAY, true)
            .option(ChannelOption.SO_KEEPALIVE, false)
            .option(ChannelOption.SO_SNDBUF, nettyConfig.getSocketSndBufSize())
            .option(ChannelOption.SO_RCVBUF, nettyConfig.getSocketRcvBufSize())
            .handler(new NettyClientInitializer(this));
        
        scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                try {
                    NettyRemotingClientAbstract.this.scanTimeoutAsyncResponseTable();
                } catch (Exception e) {
                    getLog().error("CLIENT-NETTY: scan response table exception:", e);
                }
            }
        }, 1000, 60 * 1000, TimeUnit.MILLISECONDS);
    }
    
    public void shutdown() {
        eventLoopGroupWorker.shutdownGracefully();
        scheduledExecutorService.shutdown();
        processsorExecutor.shutdown();
        callBackExecutor.shutdown();
    }
    
    @Override
    public Channel getAndCreateChannel(final String addr) {
        NettyChannelWrapper channelWrapper = channelTables.get(addr);
        if (channelWrapper != null && channelWrapper.isOK()) {
            return channelWrapper.getChannel();
        }
        try {
            if (channelTableLock.tryLock(lockTimeoutMillis, TimeUnit.MILLISECONDS)) {
                boolean createNewConnection = false;
                channelWrapper = channelTables.get(addr);
                if (channelWrapper != null) {
                    if (channelWrapper.isOK()) { // channel正常
                        return channelWrapper.getChannel();
                    } else if (!channelWrapper.getChannelFuture().isDone()) {// 正在连接，退出锁等待
                        createNewConnection = false;
                    } else { // 说明连接不成功
                        channelTables.remove(addr);
                        createNewConnection = true;
                    }
                } else {
                    createNewConnection = true;
                }
                
                if (createNewConnection) {
                    ChannelFuture channelFuture = bootstrap.connect(RemotingUtil.string2SocketAddress(addr));
                    getLog().info("CLIENT-NETTY: create a new channel is connecting to server[{}]", addr);
                    channelWrapper = new NettyChannelWrapper(channelFuture);
                    channelTables.put(addr, channelWrapper);
                }
            } else {
                getLog().error("CLIENT-NETTY: create a new channel{} timeout try to get a lock {}ms.", addr, lockTimeoutMillis);
            }
        } catch (InterruptedException e) {
            getLog().error("CLIENT-NETTY: create a new channel{} try to get a lock is interrupted.", addr);
        } finally {
            channelTableLock.unlock();
        }
        
        if (channelWrapper != null) {
            ChannelFuture channelFuture = channelWrapper.getChannelFuture();
            if (channelFuture.awaitUninterruptibly(nettyConfig.getConnectTimeoutMillis())) {
                if (channelWrapper.isOK()) {
                    getLog().info("CLIENT-NETTY: create a new channel{} is ok.", channelFuture.channel());
                    return channelWrapper.getChannel();
                } else {
                    getLog().error("CLIENT-NETTY: create a new channel{} exception:", channelFuture.channel(), channelFuture.cause());
                }
            } else {
                getLog().error("CLIENT-NETTY: create a new channel{} timeout {}ms", channelFuture.channel(), nettyConfig.getConnectTimeoutMillis());
            }
        }
        return null;
    }
    
    @Override
    public void closeChannel(String addr, Channel channel) {
        if (null == channel) {
            getLog().error("CLIENT-NETTY: close a null channel.");
            return;
        }
        try {
            if (channelTableLock.tryLock(lockTimeoutMillis, TimeUnit.MILLISECONDS)) {
                if (StringUtils.isNoneBlank(addr)) {
                    NettyChannelWrapper channelWrapper = channelTables.remove(addr);
                    if (channelWrapper == null) {
                        getLog().info("CLIENT-NETTY: close channel{} has been removed", channel);
                    } else {
                        getLog().info("CLIENT-NETTY: close channel{} ok", channel);
                        RemotingUtil.closeChannel(channel);
                    }
                } else {
                    Iterator<NettyChannelWrapper> it = channelTables.values().iterator();
                    while (it.hasNext()) {
                        NettyChannelWrapper prevChannelWrapper = it.next();
                        if (prevChannelWrapper.getChannel() == channel) {
                            it.remove();
                            RemotingUtil.closeChannel(channel);
                            getLog().info("CLIENT-NETTY: close channel{} ok", channel);
                            break;
                        }
                    }
                }
            } else {
                getLog().error("CLIENT-NETTY: close channel{} timeout try to lock {}ms", channel, lockTimeoutMillis);
            }
        } catch (InterruptedException e) {
            getLog().error("CLIENT-NETTY: close channel{} is interrupted try to get a lock.", addr);
        } finally {
            channelTableLock.unlock();
        }
    }
    
    @Override
    public void removeResponse(int opaque) {
        super.getResponseTable().remove(opaque);
    }
    
    @Override
    public INettyRemotingClient registerProcessor(int requestCode, INettyProcessor processor) {
        RpcPair<INettyProcessor, ExecutorService> pair = new RpcPair<INettyProcessor, ExecutorService>(processor, this.processsorExecutor);
        super.getProcessorTable().put(requestCode, pair);
        return this;
    }
    
    @Override
    public int getClientChannelMaxIdleTimeSeconds() {
        return nettyConfig.getClientChannelMaxIdleTimeSeconds();
    }
    
    @Override
    public ExecutorService getCallBackExecutor() {
        return callBackExecutor;
    }
    
    @Override
    public ExecutorService getProcesssorExecutor() {
        return processsorExecutor;
    }
    
    @Override
    public DefaultEventExecutorGroup getDefaultEventExecutorGroup() {
        return defaultGroup;
    }
    
}
